package com.chongba.schedule.service;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.chongba.cache.CacheService;
import com.chongba.entity.Constants;
import com.chongba.entity.Task;
import com.chongba.exception.ScheduleSystemException;
import com.chongba.exception.TaskNotExistException;
import com.chongba.schedule.conf.SystemParams;
import com.chongba.schedule.inf.TaskService;
import com.chongba.schedule.mapper.TaskInfoLogsMapper;
import com.chongba.schedule.mapper.TaskInfoMapper;
import com.chongba.schedule.pojo.TaskInfoEntity;
import com.chongba.schedule.pojo.TaskInfoLogsEntity;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskInfoMapper taskMapper;

    @Autowired
    private TaskInfoLogsMapper taskLogMapper;

    @Autowired
    private CacheService cacheService;

/*
    @Override
    public long addTask(Task task) throws ScheduleSystemException {
        try {
            //未执行任务入库
            TaskInfoEntity taskInfo = new TaskInfoEntity();
            taskInfo.setTaskType(task.getTaskType());
            taskInfo.setParameters(task.getParameters());
            taskInfo.setPriority(task.getPriority());
            taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskMapper.insert(taskInfo);

            //设置任务主键id
            task.setTaskId(taskInfo.getTaskId());

            // 记录任务日志
            TaskInfoLogsEntity taskLog = new TaskInfoLogsEntity();
            taskLog.setExecuteTime(taskInfo.getExecuteTime());
            taskLog.setPriority(taskInfo.getPriority());
            taskLog.setParameters(taskInfo.getParameters());
            taskLog.setTaskType(taskInfo.getTaskType());
            taskLog.setTaskId(taskInfo.getTaskId());
            taskLog.setVersion(1);
            taskLog.setStatus(Constants.SCHEDULED);
            taskLogMapper.insert(taskLog);
        } catch (Exception e) {
            log.warn("add task exception taskid={}",task.getTaskId());
            throw new ScheduleSystemException(e.getMessage());
        }
        return task.getTaskId();
    }*/

    private static final Logger threadLogger = LoggerFactory.getLogger("thread");

    @Resource(name = "visiableThreadPool")
    private ThreadPoolTaskExecutor threadPool;

    @Override
    @Transactional
    public long addTask(Task task) throws ScheduleSystemException {
        /**
         * addTaskToDb()
         * 成功之后添加到缓存addTaskToCache();
         */
        Future<Long> future = threadPool.submit(new Callable<Long>() {
            @Override
            public Long call() throws Exception {

                boolean success = addTaskToDb(task);
                if(success){
                    addTaskToCache(task);
                }
                return task.getTaskId();
            }
        });
        long taskId = -1;
        //获取结果
        try {
            taskId = future.get(5, TimeUnit.SECONDS);
        } catch (Exception e) {
            // log.warn("add task exception");
            threadLogger.error("add task error, msg={}",e.getMessage());
            throw new ScheduleSystemException(e);
        }
        return taskId;

    }

    private void addTaskToCache(Task task) {
        // 使用任务类型和优先级作为key
        String key = task.getTaskType()+"_"+task.getPriority();
        // 判断任务应该放入未来数据集合还是消费者队列
        if(task.getExecuteTime() <= System.currentTimeMillis()){
            cacheService.lRightPush(Constants.TOPIC+key,JSON.toJSONString(task));
        }else if(task.getExecuteTime() <= nextScheduleTime){
            //  nextScheduleTime需要定义个成员变量，然后在数据恢复方法中赋值
            cacheService.zAdd(Constants.FUTURE+key,JSON.toJSONString(task),task.getExecuteTime());
        }
        //cacheService.zAdd(Constants.DBCACHE, JSON.toJSONString(task),task.getExecuteTime());
    }

    private boolean addTaskToDb(Task task) throws ScheduleSystemException{
        boolean flag = false;
        try {
            //未执行任务入库
            TaskInfoEntity taskInfo = new TaskInfoEntity();
            taskInfo.setTaskType(task.getTaskType());
            taskInfo.setParameters(task.getParameters());
            taskInfo.setPriority(task.getPriority());
            taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
            taskMapper.insert(taskInfo);

            //设置任务主键id
            task.setTaskId(taskInfo.getTaskId());

            // int i = 1/0;

            // 记录任务日志
            TaskInfoLogsEntity taskLog = new TaskInfoLogsEntity();
            taskLog.setExecuteTime(taskInfo.getExecuteTime());
            taskLog.setPriority(taskInfo.getPriority());
            taskLog.setParameters(taskInfo.getParameters());
            taskLog.setTaskType(taskInfo.getTaskType());
            taskLog.setTaskId(taskInfo.getTaskId());
            taskLog.setVersion(1);
            taskLog.setStatus(Constants.SCHEDULED);
            taskLogMapper.insert(taskLog);

            flag = true;
        } catch (Exception e) {
            log.warn("add task exception taskid={}",task.getTaskId());
            throw new ScheduleSystemException(e.getMessage());
        }
        return flag;
    }
    @Override
    @Transactional
    public boolean cancelTask(long taskId) throws TaskNotExistException {
        boolean flag = false;
        try {
            taskMapper.deleteById(taskId);
            TaskInfoLogsEntity taskLog = taskLogMapper.selectById(taskId);
            taskLog.setStatus(Constants.CANCELLED);
            taskLogMapper.updateById(taskLog);

            flag = true;
        } catch (Exception e) {
            log.warn("task cancel exception taskid={}",taskId);
            throw new TaskNotExistException(e);
        }
        return flag;
    }

    private void removeTaskFromCache(Task task) {
        String key = task.getTaskType()+"_"+task.getPriority();

        //判断从未来数据集合中移除还是从消费者队列移除
        if(task.getExecuteTime() <= System.currentTimeMillis()){
            cacheService.lRemove(Constants.TOPIC+key,0,JSON.toJSONString(task));
        }else {
            cacheService.zRemove(Constants.FUTURE+key,JSON.toJSONString(task));
        }
        //cacheService.zRemove(Constants.DBCACHE,JSON.toJSONString(task));
    }

    private Task updateDb(long taskId,int status) throws TaskNotExistException {
        Task task = null;
        try {
            //删除任务信息表数据
            taskMapper.deleteById(taskId);
            //修改日志信息表数据为取消状态
            TaskInfoLogsEntity taskLog = taskLogMapper.selectById(taskId);
            taskLog.setStatus(status);
            taskLogMapper.updateById(taskLog);

            // 构造返回的任务对象
            task = new Task();
            /*task.setTaskId(taskLog.getTaskId());
            task.setParameters(taskLog.getParameters());*/
            // BeanUtils.copyProperties是spring框架提供的属性拷贝工具类  反射
            BeanUtils.copyProperties(taskLog,task);
            task.setExecuteTime(taskLog.getExecuteTime().getTime());
        } catch (Exception e) {
            log.warn("task cancel exception taskid={}",taskId);
            throw new TaskNotExistException(e);
        }
        return task;
    }

    /*@Override
    public long size() {
        Set<String> rangeAll = cacheService.zRangeAll(Constants.DBCACHE);
        return rangeAll.size();
   }*/
    @Override
    public long size(int type,int priority) {
        // 任务数量=未来数据集合中的任务数量+消费者队列中的数量
        String key = type+"_"+priority;
        Set<String> zRangeAll = cacheService.zRangeAll(Constants.FUTURE + key);

        Long len = cacheService.lLen(Constants.TOPIC + key);

        return zRangeAll.size() + len;
        /*Set<String> rangeAll = cacheService.zRangeAll(Constants.DBCACHE);
        return rangeAll.size();*/
    }

   /* @Override
    @Transactional
    public Task poll() throws TaskNotExistException {
        Task task = null;
        try {
            //从zset中拉取当前可执行的任务
            Set<String> byScore = cacheService.zRangeByScore(Constants.DBCACHE, 0, System.currentTimeMillis());
            if(byScore!=null && !byScore.isEmpty()){
               //获取当前可执行任务中的第一个任务
                String task_json = byScore.iterator().next();
                if(!StringUtils.isEmpty(task_json)){
                    task = JSON.parseObject(task_json,Task.class);
                    //从缓存中移除该任务
                    cacheService.zRemove(Constants.DBCACHE,task_json);
                    //更新数据库任务信息
                    updateDb(task.getTaskId(),Constants.EXECUTED);
                }
            }
        } catch (TaskNotExistException e) {
            log.warn("poll task exception");
            throw new TaskNotExistException(e);
        }
        return task;
    }*/

    @Override
    @Transactional
    public Task poll(int type,int priority) throws TaskNotExistException {

        Future<Task> future = threadPool.submit(new Callable<Task>() {
            @Override
            public Task call() throws Exception {
                Task task = null;
                //从消费者队列中获取任务
                String key = type+"_"+priority;
                String task_json = cacheService.lLeftPop(Constants.TOPIC + key);
                if(!StringUtils.isEmpty(task_json)){
                    task = JSON.parseObject(task_json,Task.class);

                    //跟新数据库信息
                    updateDb(task.getTaskId(),Constants.EXECUTED);
                }
                return task;
            }
        });

        //获取线程返回结果
        Task task = null;
        try {
            task = future.get(5,TimeUnit.SECONDS);
        } catch (Exception e) {
            // log.warn("poll task exception");
            threadLogger.error("poll task error,type={},priority={}",type,priority);
            throw new TaskNotExistException(e);
        }
        return task;
    }

    private long nextScheduleTime;

    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    @Autowired
    private SystemParams systemParams;

    @PostConstruct
    private void syncData(){
        threadPoolTaskScheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                threadLogger.info("reloadData---");
                reloadData();
            }
        },TimeUnit.MINUTES.toMillis(systemParams.getPreLoad()));

    }

    private void reloadData(){
        System.out.println("init ..............");
        // 清除缓存中原有的数据
        clearCache();
        QueryWrapper<TaskInfoEntity> wrapper = new QueryWrapper<>();
        wrapper.select("task_type","priority");
        wrapper.groupBy("task_type","priority");
        List<Map<String, Object>> maps = taskMapper.selectMaps(wrapper);

        long start = System.currentTimeMillis();

        CountDownLatch latch = new CountDownLatch(maps.size());

        //获取未来5分钟时间
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,systemParams.getPreLoad());
        nextScheduleTime = calendar.getTimeInMillis();

        for (Map<String, Object> map : maps) {

            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    long start = System.currentTimeMillis();

                    String taskType = String.valueOf(map.get("task_type"));
                    String priority = String.valueOf(map.get("priority"));
                    //查询每组内的所有任务,根据类型和优先级来查询任务集合
                    //List<TaskInfoEntity> allTaskInfo = taskMapper.queryAll(Integer.parseInt(taskType), Integer.parseInt(priority));
                    List<TaskInfoEntity> allTaskInfo = taskMapper.queryFutureTime(Integer.parseInt(taskType), Integer.parseInt(priority),calendar.getTime());
                    for (TaskInfoEntity taskInfoEntity : allTaskInfo) {
                        Task task = new Task();
                        //属性拷贝
                        BeanUtils.copyProperties(taskInfoEntity,task);
                        task.setExecuteTime(taskInfoEntity.getExecuteTime().getTime());
                        //放入缓存
                        addTaskToCache(task);
                    }

                    latch.countDown();
                    //追踪每个分组线程的信息
                    threadLogger.info("线程{},计数器{},每组恢复耗时{}",Thread.currentThread().getName(),latch.getCount(),System.currentTimeMillis()-start);
                }
            });
        }

        //阻塞主线程，等待线程返回  latch=0
        try {
            latch.await(1,TimeUnit.MINUTES);
            threadLogger.info("数据恢复完成,共耗时:"+(System.currentTimeMillis()-start)+"毫秒");
        } catch (InterruptedException e) {
            threadLogger.error("数据恢复失败,失败原因{}",e.getMessage());
        }
    }
    /*@PostConstruct
    private void syncData(){
        System.out.println("init ..............");
        // 清除缓存中原有的数据
        clearCache();
        QueryWrapper<TaskInfoEntity> wrapper = new QueryWrapper<>();
        wrapper.select("task_type","priority");
        wrapper.groupBy("task_type","priority");
        List<Map<String, Object>> maps = taskMapper.selectMaps(wrapper);

        long start = System.currentTimeMillis();

        CountDownLatch latch = new CountDownLatch(maps.size());

        //获取未来5分钟时间
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        nextScheduleTime = calendar.getTimeInMillis();

        for (Map<String, Object> map : maps) {

            threadPool.execute(new Runnable() {
                @Override
                public void run() {
                    long start = System.currentTimeMillis();

                    String taskType = String.valueOf(map.get("task_type"));
                    String priority = String.valueOf(map.get("priority"));
                    //查询每组内的所有任务,根据类型和优先级来查询任务集合
                    //List<TaskInfoEntity> allTaskInfo = taskMapper.queryAll(Integer.parseInt(taskType), Integer.parseInt(priority));
                    List<TaskInfoEntity> allTaskInfo = taskMapper.queryFutureTime(Integer.parseInt(taskType), Integer.parseInt(priority),calendar.getTime());

                    for (TaskInfoEntity taskInfoEntity : allTaskInfo) {
                        Task task = new Task();
                        //属性拷贝
                        BeanUtils.copyProperties(taskInfoEntity,task);
                        task.setExecuteTime(taskInfoEntity.getExecuteTime().getTime());
                        //放入缓存
                        addTaskToCache(task);
                    }

                    latch.countDown();
                    //追踪每个分组线程的信息
                    threadLogger.info("线程{},计数器{},每组恢复耗时{}",Thread.currentThread().getName(),latch.getCount(),System.currentTimeMillis()-start);
                }
            });
        }

        //阻塞当前线程，等待线程池线程返回  latch=0，注意该方法要在循环外执行
        try {
            latch.await(1,TimeUnit.MINUTES);
            threadLogger.info("数据恢复完成,共耗时:"+(System.currentTimeMillis()-start)+"毫秒");
        } catch (InterruptedException e) {
            threadLogger.error("数据恢复失败,失败原因{}",e.getMessage());
        }
       *//* //从数据库查询所有任务数据
        List<TaskInfoEntity> allTaskInfo = taskMapper.selectAll();
        //将任务数据存入缓存
        for (TaskInfoEntity taskInfoEntity : allTaskInfo) {
            Task task = new Task();
            //属性拷贝
            BeanUtils.copyProperties(taskInfoEntity,task);
            task.setExecuteTime(taskInfoEntity.getExecuteTime().getTime());

            //放入缓存
            addTaskToCache(task);
        }*//*
    }*/

   /* private void clearCache() {
        //移除所有的数据
        cacheService.delete(Constants.DBCACHE);
    }*/

    private void clearCache() {
        // 获取未来数据集合所有的key
        Set<String> futureKeys = cacheService.scan(Constants.FUTURE + "*");// future_*
        cacheService.delete(futureKeys);
        //获取消费者队列所有的key
        Set<String> topicKeys = cacheService.scan(Constants.TOPIC + "*");// topic_*
        cacheService.delete(topicKeys);
    }

//    @Scheduled(cron = "*/1 * * * * ?")
//    public void refresh(){
//        System.out.println("refesh "+System.currentTimeMillis()/1000);
//    }


    @Scheduled(cron = "*/1 * * * * ?")
    public void refresh(){
        System.out.println("refesh "+System.currentTimeMillis()/1000);
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                //从未来数据集合中获取所有的key
                Set<String> keys = cacheService.scan(Constants.FUTURE + "*");// future_*
                for (String key : keys) {
                    // key= future_1001_3
                    String topicKey =Constants.TOPIC+key.split(Constants.FUTURE)[1];
                    //获取当前key的任务数据集合
                    Set<String> values = cacheService.zRangeByScore(key, 0, System.currentTimeMillis());
                    //将每个任务数据添加到消费者队列并从未来数据集合中删除
                    if(!values.isEmpty()){
                   /*for (String value : values) {
                       cacheService.lRightPush(topicKey,value);// topic_1001_3
                       cacheService.zRemove(key,value);
                   }*/
                        cacheService.refreshWithPipeline(key,topicKey,values);
                        System.out.println("成功的将"+key+"定时刷新到"+topicKey);
                    }
                }
            }
        });
    }
}
